package skewreduce.framework;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import skewreduce.framework.physical.PGenerateOp;
import skewreduce.framework.physical.PhysicalOp;
import skewreduce.lib.Partition;


/**
 * compile an initial plan given driver, input, output, and partition info
 */
public class EaggPlanCompiler {
	private static Logger LOG = LoggerFactory.getLogger("PLAN");
	
//    private static PhysicalOp visit(
//            EaggPlan plan,
//            PhysicalOp beginOp,
//            PhysicalOp parent,
//            Partition partition) {
//
//        PhysicalOp newOp = null;
//
//        if ( partition.isLeaf() ) {
//            // time to add worker and partition
//            newOp = plan.createWorkOp(partition);
//            // FIXME should be generate operator
//            // FIXME if this partition should be repartitioned, schedule partition again
//            plan.addDependency(beginOp,newOp);
//            
//            // depending on criteria
//            //  if leaf is above threshold (or repartition trigger), schedule partition, followed by generate operator
//            //  else create a work operator
//        } else {
//            // create merge node
//            newOp = plan.createMergeOp(partition);
//        }
//
//        // add link to parent
//        plan.addDependency(newOp,parent);
//
//        if ( partition.hasLeft() )
//            visit(plan,beginOp,newOp,partition.getLeft());
//
//        if ( partition.hasRight() )
//            visit(plan,beginOp,newOp,partition.getRight());
//
//        return newOp;
//    }

  private static PhysicalOp visit(EaggPlan plan,
		  PhysicalOp beginOp,
		  PhysicalOp parent,
		  Partition partition) {
	  return visit(plan,beginOp,parent,partition,0.0);
  }

    private static PhysicalOp visit(
            EaggPlan plan,
            PhysicalOp beginOp,
            PhysicalOp parent,
            Partition partition,double threshold) {

        PhysicalOp newOp = null;

        if ( partition.isLeaf() ) {
            // time to add worker and partition
    		Schedule s = plan.getSchedule().get(partition.getPartitionID());
    		if ( s == null || threshold == 0.0 || s.getExpectedRuntime() < threshold ) {
                newOp = plan.createWorkOp(partition);
                plan.addDependency(beginOp,newOp);
    		} else {
    			PhysicalOp pop = plan.createPartitionOp(partition);
    			PGenerateOp gop = plan.createGenerateOp(partition);
    			gop.setPartitionOp(pop);
    			
    			// create dependency chain
    			plan.addDependency(beginOp, pop);
    			plan.addDependency(pop,gop);
    			
    			newOp = gop;
    		}
        } else {
            // create merge node
            newOp = plan.createMergeOp(partition);
        }

        // add link to parent
        plan.addDependency(newOp,parent);

        if ( partition.hasLeft() )
            visit(plan,beginOp,newOp,partition.getLeft(),threshold);

        if ( partition.hasRight() )
            visit(plan,beginOp,newOp,partition.getRight(),threshold);

        return newOp;
    }

//    private static PhysicalOp visit2(
//            EaggPlan plan,
//            PhysicalOp parent,
//            Partition partition) {
//
//        PhysicalOp newOp = null;
//        PhysicalOp partOp = null;
//
//        if ( partition.isLeaf() ) {
//            // time to add worker and partition
//            newOp = plan.createWorkOp2(partition);
//        } else {
//            // create merge node
//            newOp = plan.createMergeOp(partition);
//            if ( plan.needPartitionOp(partition) ) {
//            	partOp = plan.pushPartitionOp(partition);
//            }
//        }
//
//        // add link to parent
//        plan.addDependency(newOp,parent);
//
//        if ( partition.hasLeft() )
//            visit2(plan,newOp,partition.getLeft());
//
//        if ( partition.hasRight() )
//            visit2(plan,newOp,partition.getRight());
//        
//        if ( partOp != null )
//        	plan.popPartitionOp();
//        
//        return newOp;
//    }

    /**
     * compile physical plan from given partitioning information
     */
    public static void compileInitialPlan(EaggPlan plan,PhysicalOp op) {
        Partition root = plan.getPartitionInfo();
        if ( LOG.isDebugEnabled() ) {
        	LOG.debug(root.toLongString());
        }
        
    	plan.setRootPartitionOp(root);
        plan.setFinalizeOp(root);

        // setup task input and output
        plan.getRootPartitionOp().addInputPath(plan.getJobInputPath());
        plan.getFinalizeOp().setOutputPath(plan.getJobOutputPath());

        PhysicalOp topMerge = visit(plan,plan.getRootPartitionOp(),plan.getFinalizeOp(),root);
        plan.setFinalMergeOp(topMerge);

        plan.addDependency(op,plan.getRootPartitionOp());
        
        dumpPlan(plan);
    }
    
    /**
     * compile physical plan from given partitioning information
     */
//    public static void compileManualPlan(EaggPlan plan,PhysicalOp op) {
//        Partition root = plan.getPartitionInfo();
//        if ( LOG.isDebugEnabled() ) {
//        	LOG.debug(root.toLongString());
//        }
//        
//        // root partition is automatically set by visit2
//    	//plan.setRootPartitionOp(root);
//        plan.setFinalizeOp(root);
//
//        PhysicalOp topMerge = visit2(plan,plan.getFinalizeOp(),root);
//        plan.setFinalMergeOp(topMerge);
//
//        // trim unnecessary partition operators
//        plan.removeDanglingPartitionOp();
//
//        // configure root partitioner
//        plan.addDependency(op,plan.getRootPartitionOp());
//        
//        Configuration conf = plan.getRootPartitionOp().getExtraConf();
//        conf.setInt("partitioner.root.level", plan.getRootPartitionLevel());
//       
//        // setup task input and output
//        plan.getRootPartitionOp().addInputPath(plan.getJobInputPath());
//        plan.getFinalizeOp().setOutputPath(plan.getJobOutputPath());
//        
//        dumpPlan(plan);
//    }
    
    /**
     * compile physical plan from given partitioning information
     * this method will compile a single root partition operator 
     * rather than multiple partition operators. thus, no overhead
     * to schedule deep branches.
     */
    public static void compileManualPlan2(EaggPlan plan,PhysicalOp op) {
        Partition root = plan.getPartitionInfo();
        if ( LOG.isDebugEnabled() ) {
        	LOG.debug(root.toLongString());
        }
        
    	plan.setRootPartitionOp(root);
        plan.setFinalizeOp(root);

        PhysicalOp topMerge = visit(plan,plan.getRootPartitionOp(),plan.getFinalizeOp(),root);
        plan.setFinalMergeOp(topMerge);

        // setup task input and output
        plan.getRootPartitionOp().addInputPath(plan.getJobInputPath());
        plan.getFinalizeOp().setOutputPath(plan.getJobOutputPath());
        
        dumpPlan(plan);
    }
 
    public static void generateInitialPlan(EaggPlan plan,PhysicalOp op) {
        Partition root = plan.getPartitionInfo();
        if ( LOG.isDebugEnabled() ) {
        	LOG.debug(root.toLongString());
        }
        
        // double check whether we have to set a threshold
        double threshold = 0.0;
        if ( plan.getInitialPartitionStrategy() == EaggPlan.InitialPartitionStrategy.PARTIAL_DYNAMIC ) {
        	//threshold = plan.getSchedule().estimateWorkPhase(plan.getClusterCapacity()) * 0.5;
        	threshold = plan.getSchedule().estimateWorkPhase();
        	if ( LOG.isInfoEnabled() ) {
        		LOG.info("Setting repartitioning threshold to {}",threshold);
        	}
        }
    
        plan.setFinalizeOp(root);

        PhysicalOp topMerge = visit(plan,op,plan.getFinalizeOp(),root,threshold);
        plan.setFinalMergeOp(topMerge);

        // setup task input and output
        plan.getFinalizeOp().setOutputPath(plan.getJobOutputPath());
        
        dumpPlan(plan);
    }
    
    public static void generateIncrementalPlan(EaggPlan plan,PhysicalOp op) {
        Partition root = op.getBound();
        if ( LOG.isDebugEnabled() ) {
        	LOG.debug(root.toLongString());
        }
       
        List<PhysicalOp> downOps = plan.getDownStreamOps(op);
        if ( downOps.size() != 1 ) {
        	throw new IllegalStateException(String.format("Illegal number of downstream operators: %s",Arrays.toString(downOps.toArray())));
        }
        
        // remove dependency -- op will be a generate operator. the down stream will be a merge.
        // the revised plan should be interposed between generate and merge.
        PhysicalOp downOp = downOps.get(0);
        //plan.removeDependency(op, downOps.get(0));

        visit(plan,op,downOp,root);
        
        plan.removeDependency(op, downOp); // now we are safe to remove the dependency
        
        //dumpPlan(plan);
    }
    
    public static EaggPlan compile(EaggDriver driver,Path input,Path output)
    throws IOException {
    	Partition root = driver.getRootPartition();
    	
    	EaggPlan plan = new EaggPlan(driver,input,output,root);
    	
    	// possible initial plans
    	// * no manual plan
    	//   sample -> optimize -> partition
    	// * manual plan
    	//   optimize -> partition
    	// * no manual plan. fully dynamic
    	//   sample -> partition -> optimize
    	// * manual plan. fully dynamic
    	//   partition -> optimize
    	// * manual plan. partial dynamic
    	//   optimize -> partition -> repartition
    	
    	plan.setRootPartitionOp(root);
    	
    	if ( plan.hasManualPlan() ) {
            plan.getRootPartitionOp().addInputPath(input);
    	} else {
     		PhysicalOp sampleOp = plan.createSampleOp(root);
    		sampleOp.addInputPath(input);
	    	plan.addDependency(sampleOp,plan.getRootPartitionOp());
    	}

    	PGenerateOp genOp = plan.createGenerateOp(root);
    	genOp.setPartitionOp(plan.getRootPartitionOp());
    	plan.addDependency(plan.getRootPartitionOp(),genOp);
    	
    	dumpPlan(plan);
    	
    	return plan;
    }
    
    public static void compileRevisedPlan(EaggPlan plan,PhysicalOp op) {
		Partition context = op.getBound();
        if ( LOG.isDebugEnabled() ) {
        	LOG.debug(context.toLongString());
        }
		List<PhysicalOp> downOps = plan.getDownStreamOps(op);
		
		PhysicalOp partitionOp = plan.createPartitionOp(context);
		PhysicalOp work0   = plan.createWorkOp(context.getLeft());
		PhysicalOp work1   = plan.createWorkOp(context.getRight());
		PhysicalOp mergeOp = plan.createMergeOp(context);

        partitionOp.addInputPath( op.getInputPath() );
		
        plan.addDependency(op,partitionOp);
		plan.addDependency(partitionOp,work0);
		plan.addDependency(partitionOp,work1);
		plan.addDependency(work0,mergeOp);
		plan.addDependency(work1,mergeOp);
		
		for ( PhysicalOp downOp : downOps ) {
			plan.addDependency(mergeOp,downOp);
			plan.removeDependency(op,downOp);
		}
		
		dumpPlan(plan);
    }
    
    private static void dumpPlan(EaggPlan plan) {
    	try {
    		plan.exportDot(String.format("%s-plan.%d",plan.getDriver().getAppName(),System.currentTimeMillis()));
    	} catch ( IOException ignore ) {}
    }
}
